package com.tang.lts.core.ec.injvm;

import com.tang.lts.core.ec.EventCenter;
import com.tang.lts.core.ec.EventInfo;
import com.tang.lts.core.ec.EventSubscriber;
import com.tang.lts.core.factory.NameThreadFactory;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author Tang
 * @classname InjvmEventCenter
 * @description [  ]
 * @date 2022/4/28 15:01
 */
public class InjvmEventCenter implements EventCenter {

    private final ConcurrentHashMap<String, Set<EventSubscriber>> ecMap = new ConcurrentHashMap<>();

    private final ExecutorService executor = Executors.newFixedThreadPool(2, new NameThreadFactory("Injvm",true));

    @Override
    public void subscribe(EventSubscriber eventSubscribe, String topic) {
        Set<EventSubscriber> eventSubscribes = ecMap.get(topic);
        if (eventSubscribe == null) {
            eventSubscribes = new HashSet<>();
        }
        eventSubscribes.add(eventSubscribe);
        ecMap.put(topic, eventSubscribes);
    }

    @Override
    public void unSubscribe(EventSubscriber subscriber, String topic) {
        Set<EventSubscriber> subscribers = ecMap.get(topic);
        if (subscribers != null) {
            for (EventSubscriber eventSubscriber : subscribers) {
                if (eventSubscriber.getId().equals(subscriber.getId())) {
                    subscribers.remove(eventSubscriber);
                }
            }
        }
    }

    @Override
    public void publishSync(EventInfo eventInfo) {
        Set<EventSubscriber> subscribers = ecMap.get(eventInfo.getTopic());
        for (EventSubscriber subscriber : subscribers) {
            subscriber.getObserver()
                    .onObserver(eventInfo);
        }
    }

    @Override
    public void publishAsync(EventInfo eventInfo) {
        executor.submit(() -> {
            publishSync(eventInfo);
        });
    }
}